package udf.web;

import java.io.IOException;
import java.util.LinkedList;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import util.webSearch.BingQueryJsonHttp;
import util.webSearch.WebResult;

public class FetchBingResultsHttp extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";
	String AGENT_LIST = "/dictionary/user_agent_list.txt";
	public LinkedList<WebResult> getBingResults(String query, int max_results,
			long delay) throws IOException {
		LinkedList<WebResult> suggestions = null;
		BingQueryJsonHttp bing = new BingQueryJsonHttp();

		
		
		suggestions = bing.bingQuery(query, max_results, AGENT_LIST);

		bing = null;
		// Put to sleep this thread after the request

		try {
			Thread.sleep(delay);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw new IOException();
		}

		
		return suggestions;
	}

	/**
	 * 
	 * Input tuple contains query, max_results, delay
	 * 
	 */
	public DataBag exec(Tuple input) throws IOException {

		DataBag output = mBagFactory.newDefaultBag();

		String query = (String) input.get(0);
		String max_results_ = (String) input.get(1);
		String delay_ = (String) input.get(2);

		int max_results = Integer.valueOf(max_results_);
		long delay = Long.valueOf(delay_);

		//System.out.println("Query:" + query);
		//System.out.println("max results:" + max_results);
		LinkedList<WebResult> list=null;
		try {
			 list = getBingResults(query.trim(), max_results,
					delay);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw new IOException();
		}


		if (list == null) {
			TupleFactory mTupleFactory = TupleFactory.getInstance();
			Tuple tuple = mTupleFactory.newTuple();
			tuple.append(query);
			tuple.append("-error-");
			tuple.append("-error-");
			tuple.append("-error-");
			tuple.append(-1);
			
		
		     throw new IOException();
		    
			
			// System.out.println(query + "\t" + "-error-" + "\t" +
			// (-1));

		} else if (list.size() == 0) {
			TupleFactory mTupleFactory = TupleFactory.getInstance();
			Tuple tuple = mTupleFactory.newTuple();
			tuple.append(query);
			tuple.append("-no_results-");
			tuple.append("-no_results-");
			tuple.append("-no_results-");
			tuple.append(-2);

		} else {

			for (int i = 0; i < list.size(); i++) {

				WebResult result = list.get(i);
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append(query);
				
				
				String tmp = result.getUrl();
				
				if(tmp!=null){
					
					tmp  = tmp.replaceAll("\\s+", " ");
				}
				tuple.append(tmp);
				
				tmp = result.getTitle();
				
				if(tmp!=null){
					
					tmp  = tmp.replaceAll("\\s+", " ");
				}
				tuple.append(tmp);
				
				
				tmp = result.getSnippet();
				
				if(tmp!=null){
					
					tmp  = tmp.replaceAll("\\s+", " ");
				}
				tuple.append(tmp);
				tuple.append(result.getRank());

				output.add(tuple);
			}
		}

		return output;
		 
	}

	/**
	 * output schema will have the following format
	 * 
	 * query \t url \t title \t snippet \rank
	 * 
	 * 
	 */
	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();

			bagSchema.add(new Schema.FieldSchema("query", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("url", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("title", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("snippet", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("rank", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}
}
